package streaming.demo2

import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}

/**
  * sparkstreaming从hdfs上读取数据
  * updateStateByKey将我们累计的数据保存起来并不断的累计
  *
  */
object HdfsFileCount {
  /**
    *
    * @param inputData 输入的一个数据，包装在seq里面，seq类似于一个集合
    * @param sumData   累加之后的值 Option类型
    * @return
    *
    * 在scala当中为了避免使用null这种类型，专门包装了一个数据类型，叫做Optin，有两个子类，一个是none，一个是some
    *
    */
  def myUpdateFun(inputData: Seq[Int], sumData: Option[Int]): Option[Int] = {
    //getOrElse  用于获取option中的数据，如果没有获取到返回0
    val finalResult: Int = inputData.sum + sumData.getOrElse(0)
    //使用some保装结果值进行返回
    Some(finalResult)
  }

  def main(args: Array[String]): Unit = {
    val sparkConf: SparkConf = new SparkConf().setAppName("HdfsFileCount").setMaster("local[4]")
    val sc: SparkContext = new SparkContext(sparkConf)
    sc.setLogLevel("WARN")
    val streamingContext: StreamingContext = new StreamingContext(sc, Seconds(5))
    //使用updateStateByKey 一定要设置checkpoint用于保存历史数据
    streamingContext.checkpoint("./checkupdate2")
    val fileStream: DStream[String] = streamingContext.textFileStream("hdfs://node01:8020/stream-data")
    val wordAndOne: DStream[(String, Int)] = fileStream.flatMap(x => x.split(",")).map((_, 1))
    //通过updateStateByKey来实现数据的更新
    val key: DStream[(String, Int)] = wordAndOne.updateStateByKey(myUpdateFun)
    key.print()
        fileStream.print()
    streamingContext.start()
    streamingContext.awaitTermination()
  }
}
